#include "../StdInc.h"
#include "MsgBus.h"

namespace GSTrader
{
	namespace TradeStrategy
	{
		CMsgBus::CMsgBus(CFramework* pFramework, int nMode)
		{
			m_pMsg = NULL;
			m_bRunning = true;
			m_pFramework = pFramework;
			m_pTradePipe = new CMsgPipe;
			m_pDataPipe = new CMsgPipe;
			m_pServicePipe = new CMsgPipe;

			m_pLocalTimerQueue = NULL;
			m_pExchangeTimerQueue = NULL;
		}

		CMsgBus::~CMsgBus()
		{
			_delete(m_pServicePipe)
			_delete(m_pDataPipe)
			_delete(m_pTradePipe)
		}

		void CMsgBus::Clear()
		{
			m_pLocalTimerQueue->Clear();
			m_pExchangeTimerQueue->Clear();
			m_pTradePipe->Clear();
			m_pDataPipe->Clear();
			m_pServicePipe->Clear();

			m_lExternalQueue.clear();
		}

		void CMsgBus::AddQueue(CMsgQueue* pQueue)
		{
			if (pQueue == NULL)
				return;

			for (auto p : m_lExternalQueue)
			{
				if (pQueue == p)
					return;
			}

			m_lExternalQueue.push_back(pQueue);

			m_pDataPipe->AddMsgQueue(pQueue);
		}

		void CMsgBus::RemoveQueue(CMsgQueue* pQueue)
		{
			list<CMsgQueue*>::iterator it;
			for (it = m_lExternalQueue.begin(); it != m_lExternalQueue.end(); it++)
			{
				if (*it == pQueue)
				{
					m_lExternalQueue.erase(it);
					break;
				}
			}

			if (pQueue)
			{
				m_pDataPipe->DeleteMsgQueye(pQueue);
			}
		}

		void CMsgBus::Stop()
		{
			m_bRunning = false;
		}

		CMsgQueue* CMsgBus::GetServiceQueue()
		{
			if (m_pServicePipe->GetQueueCount() == 0)
			{
				CMsgQueue* pMsgQueue = new CMsgQueue;
				m_pServicePipe->AddMsgQueue(pMsgQueue);
			}

			return m_pServicePipe->GetQueueByIndex(0);
		}

		CMsg* CMsgBus::Pop()
		{
			if (m_pFramework->m_nMode == runRealtime || m_pFramework->m_nMode == runClient)
			{
				while (m_bRunning)
				{
					if (!m_pDataPipe->IsEmpty() && m_pMsg == NULL)
					{
						m_pMsg = m_pDataPipe->Pop();
					}

					if (!m_pLocalTimerQueue->Empty())
					{
						if (m_pLocalTimerQueue->GetTimePoint() <= m_pFramework->m_pRuntimeLocalTime->GetTime())
						{
							return m_pLocalTimerQueue->Pop();
						}
					}

					if (!m_pExchangeTimerQueue->Empty())
					{
						if (m_pMsg && m_pMsg->GetType() == msgOnNewQuoteData)
						{
							CDepthMarketData* pMarketData = (CDepthMarketData*)m_pMsg->GetData();
							if (m_pExchangeTimerQueue->GetTimePoint() <= pMarketData->GetTimePoint()) 
							{
								return m_pExchangeTimerQueue->Pop();
							}
						}
					}

					if (!m_pTradePipe->IsEmpty())
					{
						return m_pTradePipe->Pop();
					}

					if (!m_pServicePipe->IsEmpty())
					{
						return m_pServicePipe->Pop();
					}

					if (m_pMsg != NULL)
					{
						CMsg* pMsg = m_pMsg;
						m_pMsg = NULL;
						return pMsg;
					}

					ThreadSleep(1); 
				}
			}

			while (m_bRunning) 
			{
				if (!m_pDataPipe->IsEmpty() && m_pMsg == NULL)
				{
					m_pMsg = m_pDataPipe->Pop();

					CTimePoint t = m_pFramework->m_pRuntimeLocalTime->GetTime();
					if (m_pMsg->GetTime() < t)
					{
						m_pMsg->SetTime(t);
					}
				}

				if (!m_pTradePipe->IsEmpty())
				{
					return m_pTradePipe->Pop();
				}

				if (!m_pLocalTimerQueue->Empty())
				{
					if (m_pMsg && m_pLocalTimerQueue->GetTimePoint() <= m_pMsg->GetTime())
					{
						return m_pLocalTimerQueue->Pop();
					}
				}

				if (!m_pExchangeTimerQueue->Empty())
				{
					if (m_pMsg && m_pMsg->GetType() == msgOnNewQuoteData)
					{
						CDepthMarketData* pMarketData = (CDepthMarketData*)m_pMsg->GetData();
						if (m_pExchangeTimerQueue->GetTimePoint() <= pMarketData->GetTimePoint()) 
						{
							return m_pExchangeTimerQueue->Pop();
						}
					}
				}

				if (!m_pServicePipe->IsEmpty())
				{
					return m_pServicePipe->Pop();
				}

				if (m_pMsg != NULL)
				{
					CMsg* pMsg = m_pMsg;
					m_pMsg = NULL;

					for (auto p : m_lExternalQueue)
					{
						p->Push(pMsg);
					}

					return pMsg;
				}

				ThreadSleep(1);

			}

			return NULL;
		}
	}
}